package com.bw.gmall.realtime.app1.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.utils.DateFormatUtil;
import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class BaseLogApp {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        String topic = "topic_log";
        String groupId = "base_log_consumer";

        DataStreamSource<String> source = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));

//        source.print();



// TODO 4. 数据清洗，转换结构
        // 4.1 定义错误侧输出流
        OutputTag<String> dirtyStreamTag = new OutputTag<String>("dirtyStream") {
        };
        SingleOutputStreamOperator<String> cleanedStream = source.process(
                new ProcessFunction<String, String>() {
                    @Override
                    public void processElement(String jsonStr, Context ctx, Collector<String> out) throws Exception {
                        try {

                            JSON.parseObject(jsonStr);
                            out.collect(jsonStr);

                        } catch (Exception e) {
                            ctx.output(dirtyStreamTag, jsonStr);
                            System.out.println(jsonStr);
                        }
                    }
                }
        );

        cleanedStream.print();

        SingleOutputStreamOperator<JSONObject> mappedStream = cleanedStream.map(JSON::parseObject);

        // TODO 5. 新老访客状态标记修复   状态  keyby  mid

        KeyedStream<JSONObject, String> keyedStream = mappedStream
                .keyBy(r -> r.getJSONObject("common").getString("mid"));//mid相同的给共享
//

        SingleOutputStreamOperator<JSONObject> fixedStream = keyedStream.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    ValueState<String> firstViewDtState;

                    @Override
                    public void open(Configuration param) throws Exception {
                        super.open(param);

                        firstViewDtState = getRuntimeContext().getState(new ValueStateDescriptor<String>("firstViewDtState",
                                String.class));
                    }

                    @Override
                    public void processElement(JSONObject jsonObject, Context context, Collector<JSONObject> out) throws Exception {
                        String isNew=jsonObject.getJSONObject("common").getString("is_new");

                        String firstViewDt = firstViewDtState.value();

                        Long ts = jsonObject.getLong("ts");

                        String dt = DateFormatUtil.toDate(ts);
                        if ("1".equals(isNew)){
                            if (firstViewDt==null){
                                firstViewDtState.update(dt);
                            }else {
                                if (!firstViewDt.equals(dt)){
                                    isNew="0";
                                    jsonObject.getJSONObject("common").put("is_new",isNew);
                                }
                            }
                        }else {
                            if (firstViewDt==null){
                                String yesterday = DateFormatUtil.toDate(ts-1000*60*60*24);
                                firstViewDtState.update(yesterday);
                            }
                        }

                        out.collect(jsonObject);

                    }



                }
        );

        OutputTag<String> startTag = new OutputTag<String>("startTag") {
        };
        OutputTag<String> displayTag = new OutputTag<String>("displayTag") {
        };
        OutputTag<String> actionTag = new OutputTag<String>("actionTag") {
        };
        OutputTag<String> errorTag = new OutputTag<String>("errorTag") {
        };


        SingleOutputStreamOperator<String> sparseStream=fixedStream.process(
                new ProcessFunction<JSONObject, String>() {
                    @Override
                    public void processElement(JSONObject jsonObject, Context context, Collector<String> collector) throws Exception {
                        JSONObject error = jsonObject.getJSONObject("err");
                        if (error != null) {
                            context.output(errorTag, jsonObject.toJSONString());
                        }

                        jsonObject.remove("err");

                        JSONObject start=jsonObject.getJSONObject("start");
                        if(start!=null){
                            context.output(startTag, jsonObject.toJSONString());
                        }else{
                            JSONObject page = jsonObject.getJSONObject("page");
                            JSONObject common = jsonObject.getJSONObject("common");
                            Long ts = jsonObject.getLong("ts");

                            JSONArray displays = jsonObject.getJSONArray("displays");
                            if (displays != null) {
                                for (int i = 0; i < displays.size(); i++) {
                                    JSONObject display = displays.getJSONObject(i);
                                    JSONObject displayObj = new JSONObject();
                                    displayObj.put("display", display);
                                    displayObj.put("common", common);
                                    displayObj.put("page", page);
                                    displayObj.put("ts", ts);
                                    context.output(displayTag, displayObj.toJSONString());
                                }
                            }

                            JSONArray actions = jsonObject.getJSONArray("actions");
                            if(actions !=null){
                                for (int i = 0; i < actions.size(); i++) {
                                    JSONObject action = actions.getJSONObject(i);
                                    JSONObject actionObj = new JSONObject();
                                    actionObj.put("action", action);
                                    actionObj.put("common", common);
                                    actionObj.put("page", page);
                                    actionObj.put("ts", ts);
                                    context.output(actionTag, actionObj.toJSONString());
                                }
                            }
                            jsonObject.remove("displays");
                            jsonObject.remove("actions");

                            collector.collect(jsonObject.toJSONString());
                        }


                    }
                }
        );


        // TODO 7. 将数据输出到 Kafka 的不同主题
        DataStream<String> startDS = sparseStream.getSideOutput(startTag);
        DataStream<String> displayDS = sparseStream.getSideOutput(displayTag);
        DataStream<String> actionDS = sparseStream.getSideOutput(actionTag);
        DataStream<String> errorDS = sparseStream.getSideOutput(errorTag);

        String page_topic = "dwd_traffic_page_log";
        String start_topic = "dwd_traffic_start_log";
        String display_topic = "dwd_traffic_display_log";
        String action_topic = "dwd_traffic_action_log";
        String error_topic = "dwd_traffic_error_log";

        sparseStream.addSink(MyKafkaUtil.getFlinkKafkaProducer(page_topic));
        startDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(start_topic));
        displayDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(display_topic));
        actionDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(action_topic));
        errorDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(error_topic));
        env.execute();


    }
}
